package com.hadoop.seq;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapFileOutputFormat;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

//맵파일은 키값을 검색할 수 있게 색인과 함께 정렬된 시퀀스 파일. 
//물리적으로 색인이 저장된 index 파일과 데이터 내용이 저장돼 있는 data 파일로 구성.
public class MapFileCreator extends Configured implements Tool {

	public int run(String[] args) throws Exception {
		JobConf conf = new JobConf(MapFileCreator.class);
		conf.setJobName("MapFileCreator");

		//데이터를 분석할 필요가 없으므로 별도의 매퍼와 리듀서 클래스를 설정하지 않음.
		FileInputFormat.setInputPaths(conf, new Path(args[0]));
		FileOutputFormat.setOutputPath(conf, new Path(args[1]));

		// 입력데이터를 sequence file 설정 
		conf.setInputFormat(SequenceFileInputFormat.class);
		// 출력데이터를 map file 설정
		conf.setOutputFormat(MapFileOutputFormat.class);
		// 출력데이터를 운항거리로 설정
		conf.setOutputKeyClass(IntWritable.class);

		// 시퀀스 파일 압축 포맷 설정 
		SequenceFileOutputFormat.setCompressOutput(conf, true);
		SequenceFileOutputFormat
				.setOutputCompressorClass(conf, GzipCodec.class);
		SequenceFileOutputFormat.setOutputCompressionType(conf,
				CompressionType.BLOCK);

		JobClient.runJob(conf);

		return 0;
	}

	//hadoop jar packagename.jar packagename.MainClass inputfolder outputfolder
	//inputfolder의 파일들을 통해 outputfolder에 map data, map index 파일 생성
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new MapFileCreator(),
				args);
		System.out.println("## RESULT:" + res);
	}

}
